package com.itbaizhan.myconsumer1;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


//消费端限流
/*之前我们讲过MQ可以对请求进行“削峰填谷”，即通过消费端限流的
方式限制消息的拉取速度，达到保护消费端的目的*/
//消费者监听队列
@Component
public class OocConsumer {
   // @RabbitListener(queues = "my_queue")
    public void listenMessage(Message message, Channel channel) throws InterruptedException, IOException {
        //消息投递序号，消息每一次投递该值都会+1
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try{
            //获取消息
            System.out.println(new String(message.getBody()));
            //模拟业务，处理一条消息要2秒
            Thread.sleep(2000);
            //签收消息
            /*签收消息
             * 参数1： 消息投递序号
             * 参数2：是否一次可以签收多条消息*/
            channel.basicAck(deliveryTag,true);
        }catch (Exception e){
            System.out.println("消息发送失败！");
            Thread.sleep(2000);
            /*
             * 参数1： 消息投递序号
             * 参数2：是否一次可以拒签多条消息
             * 参数3：拒签后是否返回队列*/
            channel.basicNack(deliveryTag,true,true);

        }
    }
}

/*生产者批量发送消息

@Test
public void testSendBatch() {
  // 发送十条消息
  for (int i = 0; i < 10; i++) {
    rabbitTemplate.convertAndSend("my_topic_exchange", "my_routing", "send message..."+i);
   }
}
消费端配置限流机制

spring:
  rabbitmq:
   host: 192.168.0.162
   port: 5672
   username: itbaizhan
   password: itbaizhan
   virtual-host: /
   listener:
    simple:
    # 限流机制必须开启手动签收
     acknowledge-mode: manual
    # 消费端最多拉取5条消息消费，签收后不满5条才会继续拉取消息。
     prefetch: 5*/